/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.ulss.data.trans.worker;

import cn.ac.iie.ulss.data.trans.commons.RuntimeEnv;
import cn.ac.iie.ulss.data.trans.util.Meta;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.IOException;
import java.util.Random;
import org.apache.http.HttpResponse;
import org.apache.http.client.HttpClient;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.InputStreamEntity;
import org.apache.http.impl.client.DefaultHttpClient;
import org.apache.avro.Protocol;
import org.apache.avro.Schema;
import org.apache.log4j.Logger;
import org.apache.avro.generic.GenericArray;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.BinaryDecoder;
import org.apache.avro.io.DatumReader;
import org.apache.avro.io.DecoderFactory;
import org.apache.http.client.ClientProtocolException;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.config.ConnectionConfig;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;

/**
 *
 * @author Li Mingyang
 */
public class HttpSender extends Thread {

    static Logger logger = Logger.getLogger(HttpSender.class.getName());

    private Meta meta = null;

    public HttpSender(Meta meta) {
        this.meta = meta;
    }

    @Override
    public void run() {
        long startTime = 0l;
        long endTime = 0l;
        Protocol protocol = null;
        try {
            protocol = Protocol.parse(new File("docs.json"));
        } catch (IOException ex) {
            logger.error(ex.getMessage(), ex);
        }
        final Schema docsSchema = protocol.getType("docs");
        final DatumReader<GenericRecord> docsReader = new GenericDatumReader(docsSchema);
        String[] ips = meta.getDestinationParms().split("[;]");
        PoolingHttpClientConnectionManager cm = new PoolingHttpClientConnectionManager();
        cm.setMaxTotal(30);
        cm.setDefaultMaxPerRoute(30);
        cm.setDefaultConnectionConfig(ConnectionConfig.DEFAULT);
        CloseableHttpClient httpClient = HttpClients.custom().setConnectionManager(cm).build();
        while (true) {
            byte[] data = (byte[]) RuntimeEnv.TOPIC_TO_DATAPOOL.get(meta.getDataName()).poll();
            if (data == null) {
                logger.info("the ConcurrentLinkedQueue for " + meta.getDataName() + " is empty and wait.");
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    logger.error(e.getMessage(), e);
                }
            } else {
                ByteArrayInputStream docsbis = new ByteArrayInputStream(data);
                BinaryDecoder docsbd = new DecoderFactory().binaryDecoder(docsbis, null);
                GenericRecord docsRecord = new GenericData.Record(docsSchema);
                try {
                    docsReader.read(docsRecord, docsbd);
                } catch (IOException ex) {
                    logger.error(ex.getMessage(), ex);
                }
                GenericArray docSet = (GenericArray) docsRecord.get("doc_set");
                boolean flag = true;
                Random rand = new Random();
                while (flag) {
                    String url = "http://".concat(ips[rand.nextInt(ips.length)]);
                    HttpPost httppost = new HttpPost(url);
                    RequestConfig requestConfig = RequestConfig.custom().setSocketTimeout(3 * 1000).setConnectTimeout(5000).build();
                    httppost.setConfig(requestConfig);
                    InputStreamEntity reqEntity = new InputStreamEntity(new ByteArrayInputStream(data), -1);
                    reqEntity.setContentType("binary/octet-stream");
                    reqEntity.setChunked(true);
                    httppost.setEntity(reqEntity);
                    try {
                        startTime = System.currentTimeMillis();
                        HttpResponse response  = httpClient.execute(httppost);
                        endTime = System.currentTimeMillis();
                        try {
                            if (response.getStatusLine().getStatusCode() == 200) {
                                ByteArrayOutputStream out = new ByteArrayOutputStream();
                                response.getEntity().writeTo(out);
                                String resonseEn = new String(out.toByteArray());
                                if ("-1".equals(resonseEn.split("[\n]")[0])) {
                                    logger.error("send " + docSet.size() + " messages for " + meta.getDataName() + " to the " + url + " failed because the service return a wrong infomation " + resonseEn + " use " + (endTime - startTime) + " ms");
                                } else {
                                    logger.info("send " + docSet.size() + " records and size " + data.length + " bytes,use " + (endTime - startTime) + " ms for " + meta.getDataName() + ":" + response.getStatusLine());
                                    flag = false;
                                }
                            } else {
                                logger.info("send " + docSet.size() + " messages for " + meta.getDataName() + " to the " + url + " failed because " + response.getStatusLine() + " use " + (endTime - startTime) + " ms");
                            }
                        } finally {
                            httppost.releaseConnection();
                        }
                    } catch (ClientProtocolException ex) {
                        logger.info("HttpPool is full");
                        try {
                            Thread.sleep(1000);
                        } catch (InterruptedException ex1) {
                        }
                    } catch (IOException ex) {
                        endTime = System.currentTimeMillis();
                        logger.info("send " + docSet.size() + " messages for " + meta.getDataName() + " to the " + url + " failed use " + (endTime - startTime) + " ms because " + ex, ex);
                    } catch (Exception ex) {
                        endTime = System.currentTimeMillis();
                        logger.info("send " + docSet.size() + " messages for " + meta.getDataName() + " to the " + url + " failed use " + (endTime - startTime) + " ms because " + ex, ex);
                    }
                }
            }
        }
    }
}
